{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(4)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"movielens\" + os.sep + \"ml-100k\" + os.sep\n",
    "\n",
    "RATING_FILE = \"u.data\";\n",
    "USER_FILE = \"u.user\";\n",
    "ITEM_FILE = \"u.item\";\n",
    "RATING_TRAIN_FILE = \"ua.base\";\n",
    "RATING_TEST_FILE = \"ua.test\";\n",
    "\n",
    "USER_COL = \"user_id\";\n",
    "ITEM_COL = \"item_id\";\n",
    "RATING_COL = \"rating\";\n",
    "RECOMM_COL = \"recomm\";\n",
    "\n",
    "RATING_SCHEMA_STRING = \"user_id long, item_id long, rating float, ts long\";\n",
    "\n",
    "USER_SCHEMA_STRING = \"user_id long, age int, gender string, occupation string, zip_code string\";\n",
    "\n",
    "ITEM_SCHEMA_STRING = \"item_id long, title string, \"\\\n",
    "    + \"release_date string, video_release_date string, imdb_url string, \"\\\n",
    "    + \"unknown int, action int, adventure int, animation int, \"\\\n",
    "    + \"children int, comedy int, crime int, documentary int, drama int, \"\\\n",
    "    + \"fantasy int, film_noir int, horror int, musical int, mystery int, \"\\\n",
    "    + \"romance int, sci_fi int, thriller int, war int, western int\";\n",
    "\n",
    "\n",
    "def getSourceRatings() :\n",
    "    return TsvSourceBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + RATING_FILE)\\\n",
    "            .setSchemaStr(RATING_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getStreamSourceRatings() :\n",
    "    return TsvSourceStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + RATING_FILE)\\\n",
    "            .setSchemaStr(RATING_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getSourceUsers() :\n",
    "    return CsvSourceBatchOp()\\\n",
    "            .setFieldDelimiter(\"|\")\\\n",
    "            .setFilePath(DATA_DIR + USER_FILE)\\\n",
    "            .setSchemaStr(USER_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getSourceItems() :\n",
    "    return CsvSourceBatchOp()\\\n",
    "            .setFieldDelimiter(\"|\")\\\n",
    "            .setFilePath(DATA_DIR + ITEM_FILE)\\\n",
    "            .setSchemaStr(ITEM_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getStreamSourceItems() :\n",
    "    return CsvSourceStreamOp()\\\n",
    "            .setFieldDelimiter(\"|\")\\\n",
    "            .setFilePath(DATA_DIR + ITEM_FILE)\\\n",
    "            .setSchemaStr(ITEM_SCHEMA_STRING);\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1\n",
    "\n",
    "ratings = getSourceRatings();\n",
    "users = getSourceUsers();\n",
    "items = getSourceItems();\n",
    "\n",
    "ratings.registerTableName(\"ratings\");\n",
    "items.registerTableName(\"items\");\n",
    "users.registerTableName(\"users\");\n",
    "\n",
    "BatchOperator.sqlQuery(\n",
    "    \"SELECT title, cnt, avg_rating\"\n",
    "    + \" FROM ( SELECT item_id, COUNT(*) AS cnt, AVG(rating) AS avg_rating\"\n",
    "    + \"        FROM ratings \"\n",
    "    + \"        GROUP BY item_id \"\n",
    "    + \"        ORDER BY cnt DESC LIMIT 10 \"\n",
    "    + \"      ) AS t\"\n",
    "    + \" JOIN items\"\n",
    "    + \" ON t.item_id=items.item_id\"\n",
    "    + \" ORDER BY cnt DESC\"\n",
    ").print();\n",
    "\n",
    "\n",
    "BatchOperator.sqlQuery(\n",
    "    \"SELECT title, cnt, m_rating, f_rating, ABS(m_rating - f_rating) AS diff_rating\"\n",
    "    + \" FROM ( SELECT item_id, COUNT(rating) AS cnt, \"\n",
    "    + \"               AVG(CASE WHEN gender='M' THEN rating ELSE NULL END) AS m_rating, \"\n",
    "    + \"               AVG(CASE WHEN gender='F' THEN rating ELSE NULL END) AS f_rating \"\n",
    "    + \"        FROM (SELECT item_id, rating, gender FROM ratings \"\n",
    "    + \"                     JOIN users ON ratings.user_id=users.user_id)\"\n",
    "    + \"        GROUP BY item_id \"\n",
    "    + \"      ) AS t\"\n",
    "    + \" JOIN items\"\n",
    "    + \" ON t.item_id=items.item_id\"\n",
    "    + \" ORDER BY diff_rating DESC LIMIT 10\"\n",
    ").print();\n",
    "\n",
    "BatchOperator.sqlQuery(\n",
    "    \"SELECT title, cnt, m_rating, f_rating, ABS(m_rating - f_rating) AS diff_rating\"\n",
    "    + \" FROM ( SELECT item_id, COUNT(rating) AS cnt, \"\n",
    "    + \"               AVG(CASE WHEN gender='M' THEN rating ELSE NULL END) AS m_rating, \"\n",
    "    + \"               AVG(CASE WHEN gender='F' THEN rating ELSE NULL END) AS f_rating \"\n",
    "    + \"        FROM (SELECT item_id, rating, gender FROM ratings \"\n",
    "    + \"                     JOIN users ON ratings.user_id=users.user_id)\"\n",
    "    + \"        GROUP BY item_id \"\n",
    "    + \"        HAVING COUNT(rating)>=50 \"\n",
    "    + \"      ) AS t\"\n",
    "    + \" JOIN items\"\n",
    "    + \" ON t.item_id=items.item_id\"\n",
    "    + \" ORDER BY diff_rating DESC LIMIT 10\"\n",
    ").print();\n",
    "\n",
    "\n",
    "import datetime\n",
    "\n",
    "@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.TIMESTAMP(3))\n",
    "def from_unix_timestamp(ts):\n",
    "    return datetime.datetime.fromtimestamp(ts)\n",
    "\n",
    "BatchOperator.registerFunction(\"from_unix_timestamp\", from_unix_timestamp);\n",
    "\n",
    "BatchOperator.sqlQuery(\n",
    "    \"SELECT MIN(dt) AS min_dt, MAX(dt) AS max_dt \"\n",
    "    + \" FROM ( SELECT from_unix_timestamp(ts) AS dt, 1 AS grp FROM ratings) \"\n",
    "    + \" GROUP BY grp \"\n",
    ").print();\n",
    "\n",
    "ratings\\\n",
    "    .select(\"from_unix_timestamp(ts) AS dt, 1 AS grp\")\\\n",
    "    .groupBy(\"grp\",\"MIN(dt) AS min_dt, MAX(dt) AS max_dt\")\\\n",
    "    .print();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_1\n",
    "\n",
    "ratings = getSourceRatings();\n",
    "users = getSourceUsers();\n",
    "\n",
    "ratings_select = ratings.select(\"user_id, item_id AS movie_id\");\n",
    "ratings_select.firstN(5).print();\n",
    "\n",
    "ratings.select(\"user_id, item_id AS movie_id\").firstN(5).print();\n",
    "\n",
    "ratings_select = ratings.select(\"*\");\n",
    "ratings_select.firstN(5).print();\n",
    "\n",
    "ratings.alias(\"f1,f2,f3,f4\").firstN(5).print();\n",
    "\n",
    "ratings.filter(\"rating > 3\").firstN(5).print();\n",
    "ratings.where(\"rating > 3\").firstN(5).print();\n",
    "\n",
    "users.select(\"gender\").distinct().print();\n",
    "\n",
    "users.groupBy(\"gender\", \"gender, COUNT(*) AS cnt\").print();\n",
    "\n",
    "users.orderBy(\"age\", 5).print();\n",
    "users.orderBy(\"age\", offset=1, fetch=3).print();\n",
    "\n",
    "users.orderBy(\"age\", 5, order = 'desc').print();\n",
    "users.orderBy(\"age\", offset=1, fetch=3, order = 'desc').print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_2\n",
    "\n",
    "ratings = getSourceRatings();\n",
    "items = getSourceItems();\n",
    "\n",
    "left_ratings = ratings\\\n",
    "    .filter(\"user_id<3 AND item_id<4\")\\\n",
    "    .select(\"user_id, item_id, rating\");\n",
    "\n",
    "right_movies = items\\\n",
    "    .select(\"item_id AS movie_id, title\")\\\n",
    "    .filter(\"movie_id < 6 AND MOD(movie_id, 2) = 1\");\n",
    "\n",
    "print(\"# left_ratings #\");\n",
    "left_ratings.print();\n",
    "print(\"\\n# right_movies #\");\n",
    "right_movies.print();\n",
    "\n",
    "print(\"# JOIN #\");\n",
    "JoinBatchOp()\\\n",
    "    .setJoinPredicate(\"item_id = movie_id\")\\\n",
    "    .setSelectClause(\"user_id, item_id, title, rating\")\\\n",
    "    .linkFrom(left_ratings, right_movies)\\\n",
    "    .print();\n",
    "\n",
    "print(\"\\n# LEFT OUTER JOIN #\");\n",
    "LeftOuterJoinBatchOp()\\\n",
    "    .setJoinPredicate(\"item_id = movie_id\")\\\n",
    "    .setSelectClause(\"user_id, item_id, title, rating\")\\\n",
    "    .linkFrom(left_ratings, right_movies)\\\n",
    "    .print();\n",
    "\n",
    "print(\"\\n# RIGHT OUTER JOIN #\");\n",
    "RightOuterJoinBatchOp()\\\n",
    "    .setJoinPredicate(\"item_id = movie_id\")\\\n",
    "    .setSelectClause(\"user_id, item_id, title, rating\")\\\n",
    "    .linkFrom(left_ratings, right_movies)\\\n",
    "    .print();\n",
    "\n",
    "print(\"\\n# FULL OUTER JOIN #\");\n",
    "FullOuterJoinBatchOp()\\\n",
    "    .setJoinPredicate(\"item_id = movie_id\")\\\n",
    "    .setSelectClause(\"user_id, item_id, title, rating\")\\\n",
    "    .linkFrom(left_ratings, right_movies)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_3\n",
    "\n",
    "users = getSourceUsers();\n",
    "\n",
    "users_1_4 = users.filter(\"user_id<5\");\n",
    "print(\"# users_1_4 #\");\n",
    "users_1_4.print();\n",
    "\n",
    "users_3_6 = users.filter(\"user_id>2 AND user_id<7\");\n",
    "print(\"\\n# users_3_6 #\");\n",
    "users_3_6.print();\n",
    "\n",
    "UnionAllBatchOp().linkFrom(users_1_4, users_3_6).print();\n",
    "\n",
    "UnionBatchOp().linkFrom(users_1_4, users_3_6).print();\n",
    "\n",
    "IntersectBatchOp().linkFrom(users_1_4, users_3_6).print();\n",
    "\n",
    "IntersectAllBatchOp()\\\n",
    "    .linkFrom(\n",
    "        UnionAllBatchOp().linkFrom(users_1_4, users_1_4),\n",
    "        UnionAllBatchOp().linkFrom(users_1_4, users_3_6)\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "MinusBatchOp().linkFrom(users_1_4, users_3_6).print();\n",
    "\n",
    "MinusAllBatchOp()\\\n",
    "    .linkFrom(\n",
    "        UnionAllBatchOp().linkFrom(users_1_4, users_1_4),\n",
    "        UnionAllBatchOp().linkFrom(users_1_4, users_3_6)\n",
    "    )\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
